package spark.work

//import breeze.linalg.{max, sum}
import java.math.BigDecimal

import com.kongming.kmdm.common.db.bean.DataSourceBean
import com.kongming.kmdm.model.KmDataColumnType
import org.apache.spark.sql._
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer
import scala.util.Try
//import org.apache.spark.sql.functions._

/**
  * Created by liuwei on 2017/11/8.
  */
object RowToColumn3 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("RowToColumnTest").setMaster("local[8]")
    val sc = new SparkContext(sparkConf)
    val spark = SparkSession.builder

    val ss = spark.getOrCreate();

    val df = ss.createDataFrame(Seq(
      ("张三", "女", "数学", 60, 15, "张老师"),
      ("张三", "女", "语文", 70, 12, "李老师"),
      ("张三", "女", "物理", 50, 22, "张老师"),
      ("王二", "男", "数学", 50, 16, "赵老师"),
      ("王二", "男", "语文", 60, 20, "孙老师"),
      ("王二", "男", "历史", 50, 22, "李老师")
    )).toDF("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    df.createOrReplaceTempView("test")
    val title = Seq("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    val rowLabel = Seq("姓名")
    val columnLabel = List("科目", "监考老师")
    val valueLabel = Seq("成绩", "排名")
    //    val keyArr = ArrayBuffer.empty[String]
    val keyColumnArr = df.select(columnLabel.head, columnLabel.tail: _*).distinct().collect().map(_.get(0).toString)
//    val columnValueArr = columnLabel.map(column => {
//      val value = df.select(column).distinct().rdd.collect().map(_.get(0).toString)
//      (column, value)
//    }
//    )
    val schemaSeq = df.schema.map(_.name).toSeq
    val columnIdxMap = columnLabel.map(columnLabelVal => (columnLabelVal,schemaSeq.indexOf(columnLabelVal))).toMap
    val valueIdxMap = valueLabel.map(valueLabelVal => (valueLabelVal,schemaSeq.indexOf(valueLabelVal))).toMap

    val columnValuesSeq = columnLabel.map(column => {
      //??
      val values = df.select(column).distinct().rdd.collect().map(row=>{
        if(row.isNullAt(0)) {
          "eleRowToColNull"
        }
        else {
          row.get(0)
        }
      }).toSeq

      if(0==values.size) {
        (column,Seq("eleRowToColNull"))
      }
      else {
        (column,values)
      }
    })

    val colLabelValueIndexSeq = columnValuesSeq.foldLeft(Seq(1L)){
      case (indexSeq,(column,values))=>
        indexSeq++Seq(indexSeq.last*values.size)
    }

    var total = 1
//    val columnValueMap = columnLabel.map(column => {
//      val value = df.select(column).distinct().rdd.collect().map(_.get(0).toString)
////      if (!value.isEmpty) {
////        total = total * value.length
////      }
//      (column, value)
//    }
//    ).foreach{case (column, value)=>{
//            if (!value.isEmpty) {
//              total = total * value.length
//            }
//    }}.toMap
//    val columnValueKeySeq = columnValueMap.keys.toSeq

    val arr = ArrayBuffer.empty[String]
    rowLabel.foreach(arr += _)
    //    arr += rowLabel:_*
    //    val columnArr = ArrayBuffer.empty[String]
    val keyArr = arr.clone()
    columnLabel.foreach(keyArr += _)
    //    val keyRowArr = df.select(rowLabel).distinct().collect().map(_.get(0).toString)

    for (keyColumn <- keyColumnArr)
      for (value <- valueLabel) {
        val sb = new StringBuilder()
        sb.append(columnLabel(0))
        sb.append("(")
        sb.append(keyColumn)
        sb.append(")_value(")
        sb.append(value)
        sb.append(")")
        arr += sb.toString()
      }
    val groupBy = df.rdd.groupBy(row => for (rowLable <- rowLabel) yield row.get(title.indexOf(rowLable)))

    df.rdd.map(row=>{
      val userKey = row.get(0)
      val columnKey = columnLabel.map(columnLabelVal => {
        val columnLabelUserVal = if(row.isNullAt(columnIdxMap.get(columnLabelVal).get)) {
          "eleRowToColNull"
        }else {
          row.get(columnIdxMap.get(columnLabelVal).get)
        }
      }).mkString(":")
      val valuesSeq = valueLabel.map(valueLabelVal =>
        row.get(columnIdxMap.get(valueLabelVal).get))
      (userKey,Tuple2(columnKey,valuesSeq))
    }
  ).reduceByKey{
      case (colKeyAndValSeq1,colKeyAndValSeq2)=>{
        colKeyAndValSeq1
      }
    }

//    def seq(a:Row, b:Iterable[Row]) : Row ={
//      var res = new ArrayBuffer[Any]
//      b.foreach(row=>{
//        res += row.get(3) + ""
//        res += row.get(4) + ""
//      })
//      Row.fromSeq(res)
//    }
//    def comb(a:Row, b:Row) : Row ={
//  valueLabel.map(a.getAs(_))
//  columnLabel.map(columnIdx=>{
//    val currentColValue = a.getAs(columnIdx)
//    val idx = columnValueKeySeq.indexOf(columnIdx)
//  })
//    }

//    val res = groupBy.aggregateByKey(Row.fromSeq(Seq{}))(seq, comb)
//    res.foreach(
//      println(_)
//    )

  }

  def getColLabelIdx(columnValuesSeq:Seq[(String,Seq[Any])],colLabelValueIndexSeq:Seq[Long],columnKeySeq:Seq[Any]): Long = {
    columnKeySeq.zipWithIndex.map{
      case (colVal,colIdx) => colLabelValueIndexSeq.apply(colIdx)*columnValuesSeq.apply(colIdx)._2.indexOf(colVal)
    }.reduce(_+_)
  }

}
